//++
//
// Copyright (c) Microsoft Corporation.  All rights reserved.
//
// Module Name:
//
//  DispatcherProxy.cs
//    
// Abstract:
//
//  This module implements the DispatcherProxy class. The DispatcherProxy is used for sending notifications to the Dispatcher and 
//  receiving responses to those notifications back from the Dispatcher. This class makes use of three MSMQs; one for sending 
//  requests to the Dispatcher, one for receiving acknowledgements back from the Dispacther (positive and negative) and one 
//  administration queue that is used for handling failures.
//  
//--

using System;
using System.Diagnostics;
using System.Messaging;
using System.Threading;

using Microsoft.OfficeCommunicationsServer.Applications.Common;

namespace Microsoft.EnhancedReminders.Scheduler
{
    /// <summary>
    /// Interface to DispactherProxy. This is primarily needed for implementing a mock DispatcherProxy for the unit tests.
    /// </summary>
    public interface IDispatcherProxy
    {
        /// <summary>
        /// Send a notification to the outbound message queue.
        /// </summary>
        /// <param name="notificationItem">Notification to be sent</param>
        void SendNotification(NotificationItem notificationItem);
    }

    /// <summary>
    /// The DispatcherProxy is used for sending notifications to the Dispatcher and receiving responses to those notifications 
    /// back from the Dispatcher. All communication with the Dispatcher is via MSMQ.
    /// </summary>
    public sealed class DispatcherProxy : IDispatcherProxy, IDisposable
    {
        // Message queues
        private MessageQueue _sendQueue;
        private MessageQueue _responseQueue;
        private MessageQueue _administrationQueue;

        // State variables
        private bool _stopping;
        private int _refCount;

        // Wait event used when the service shuts down
        private ManualResetEvent _stoppedEvent;

        // Time period after which a notification is considered out-of-date
        private TimeSpan _sendCutoffPeriod;

        // Used to synchronize state changes
        private readonly object _syncRoot = new object();

        /// <summary>
        /// ctor.
        /// </summary>
        /// <param name="sendQueuePath">Path for the queue that sends requests to the Dispatcher</param>
        /// <param name="responseQueuePath">Path to the queue that receives responses back from the Dispatcher</param>
        /// <param name="administrationQueuePath">Path to the administration queue</param>
        /// <param name="sendCutoffPeriod">Cut-off period after which notifications will not be sent</param>
        public DispatcherProxy(string sendQueuePath, string responseQueuePath, string administrationQueuePath, TimeSpan sendCutoffPeriod)
        {
            try
            {
                _sendQueue = new MessageQueue(sendQueuePath);
            }
            catch (MessageQueueException ex)
            {
                // Log the error and then re-throw to ensure that the service doesn't start
                OperationalLogs.LogMessageQueueFatalError(sendQueuePath, ex);
                throw;
            }

            try
            {
                _responseQueue = new MessageQueue(responseQueuePath);
            }
            catch (MessageQueueException ex)
            {
                // Log the error and then re-throw to ensure that the service doesn't start
                OperationalLogs.LogMessageQueueFatalError(responseQueuePath, ex);
                throw;
            }

            try
            {
                _administrationQueue = new MessageQueue(administrationQueuePath);
            }
            catch (MessageQueueException ex)
            {
                // Log the error and then re-throw to ensure that the service doesn't start
                OperationalLogs.LogMessageQueueFatalError(administrationQueuePath, ex);
                throw;
            }

            _sendCutoffPeriod = sendCutoffPeriod;
        }

        /// <summary>
        /// Start receiving messages from the repsonse and administration queues.
        /// </summary>
        /// <remarks>
        /// This method should only be called when the service starts.        
        /// </remarks>
        public void Open()
        {
            // Start receiving response messages
            _responseQueue.BeginReceive(TimeSpan.FromMinutes(1), this, OnResponseReceived);

            // Start receiving administration messages
            _administrationQueue.BeginReceive(TimeSpan.FromMinutes(1), this, OnAdministrationReceived);
        }

        /// <summary>
        /// Waits for any active send or recieve operation to complete and then shuts down the message queues. The wait is necessary 
        /// to prevent leaving the MSMQ queues in a bad state and to make sure that any acknowledgement removed from the queue is
        /// processed.
        /// </summary>
        public void Dispose()
        {
            lock (_syncRoot)
            {
                Debug.Assert(!_stopping && _stoppedEvent == null, "Stop should not be called multiple times");

                // Prevent any new send or receive operations from being attempted
                _stopping = true;

                if (_refCount > 0)
                {
                    // There are active send and/or receive operations in progress so need to create a wait handle 
                    // that will be signalled when the operations complete
                    _stoppedEvent = new ManualResetEvent(false);
                }
            }

            if (_stoppedEvent != null)
            {
                // Wait for all active MSMQ-related operations to complete before proceeding. An infinite wait is
                // okay here because the service shutdown has a finite timeout which if exceeded will result in
                // the process shutting down.
                _stoppedEvent.WaitOne();
                _stoppedEvent.Close();
            }

            Debug.Assert(_refCount == 0, "No MSMQ operations should be active at this point");

            // Dispose of message queues
            _sendQueue.Dispose();
            _sendQueue = null;
            _responseQueue.Dispose();
            _responseQueue = null;
            _administrationQueue.Dispose();
            _administrationQueue = null;
        }

        /// <summary>
        /// Send a notification to the outbound message queue.
        /// </summary>
        /// <param name="notificationItem">Notification to be sent</param>
        public void SendNotification(NotificationItem notificationItem)
        {
            Debug.Assert(notificationItem != null, "Notification to be sent is not specified");

            // Calculate the time remaining before this notification expires
            DateTime cutoffTime = DateTime.UtcNow - _sendCutoffPeriod;
            TimeSpan timeRemainingToBeReceived = notificationItem.NotificationTime - cutoffTime;

            if (timeRemainingToBeReceived <= TimeSpan.Zero)
            {
                Logger.WriteTraceWarning(notificationItem.Context, "Failed to deliver notification because it was out-of-date. Notification: {0}", notificationItem);
                RaiseNotificationCompleted(new NotificationResult(notificationItem.Context, notificationItem.Id, NotificationResultStatus.Error, NotificationChannelType.None));
                return;
            }

            // This ensures that any custom channels are resolved
            RaiseNotificationReady(notificationItem);

            if (!notificationItem.IsEnabled)
            {
                // There are no enabled channels so there's no point trying to deliver this notification
                Logger.WriteTraceWarning(notificationItem.Context, "Failed to deliver notification because there are no enabled channels. Notification: {0}", notificationItem);
                RaiseNotificationCompleted(new NotificationResult(notificationItem.Context, notificationItem.Id, NotificationResultStatus.Error, NotificationChannelType.None));
                return;
            }

            Message message = new Message(notificationItem, new BinaryMessageFormatter());
            message.TimeToBeReceived = timeRemainingToBeReceived;
            message.ResponseQueue = _responseQueue;
            message.AdministrationQueue = _administrationQueue;
            message.AcknowledgeType = AcknowledgeTypes.NegativeReceive | AcknowledgeTypes.NotAcknowledgeReachQueue | AcknowledgeTypes.NotAcknowledgeReceive;

            // Ensure that the service isn't shutdown until this operation completes
            if (!IncrementRefCount())
            {
                // Don't send notification because the service is stopping
                return;
            }

            try
            {
                _sendQueue.Send(message);
            }
            catch (MessageQueueException ex)
            {
                Logger.WriteTraceError(notificationItem.Context, ex, "Failed to queue notification: {0}", notificationItem);
                OperationalLogs.LogMessageQueueError(_sendQueue.Path, ex);
                RaiseNotificationCompleted(new NotificationResult(notificationItem.Context, notificationItem.Id, NotificationResultStatus.Error, NotificationChannelType.None));
                return;
            }
            finally
            {
                DecrementRefCount();
            }

            Logger.WriteTraceInformation(notificationItem.Context, "Notification queued: {0}", notificationItem);
        }

        /// <summary>
        /// Raised when a notification is successfully acknowledged by the recipient.
        /// </summary>
        public event EventHandler<NotificationCompletedEventArgs> NotificationCompleted;

        /// <summary>
        /// Raised when a notification is about to be queued for delivery.
        /// </summary>
        public event EventHandler<NotificationReadyEventArgs> NotificationReady;

        /// <summary>
        /// Callback when a response message is available.
        /// </summary>
        /// <param name="asyncResult">Context of async operation</param>
        private void OnResponseReceived(IAsyncResult asyncResult)
        {
            // Ensure that the service isn't shutdown until this operation completes
            if (!IncrementRefCount())
            {
                // Don't receive the message from the response queue because the service is stopping
                return;
            }

            Message msg = null;

            try
            {
                // Extract the response message
                msg = _responseQueue.EndReceive(asyncResult);
                Debug.Assert(msg != null, "Received message should never be null");
            }
            catch (MessageQueueException ex)
            {
                // Ignore errors due to an empty queue as these are expected if this callback is invoked because of a timeout.
                if (ex.MessageQueueErrorCode != MessageQueueErrorCode.MessageNotFound && ex.MessageQueueErrorCode != MessageQueueErrorCode.IOTimeout)
                {
                    OperationalLogs.LogMessageQueueError(_responseQueue.Path, ex);
                }
            }

            ProcessResponseMessage(msg);

            // Start listening for more response messages
            _responseQueue.BeginReceive(TimeSpan.FromMinutes(1), this, OnResponseReceived);

            DecrementRefCount();
        }

        /// <summary>
        /// Processes a response message.
        /// </summary>
        /// <param name="msg">Response message</param>
        private void ProcessResponseMessage(Message msg)
        {
            if (msg == null)
            {
                return;
            }

            msg.Formatter = new BinaryMessageFormatter();

            NotificationResult result = msg.Body as NotificationResult;

            if (result != null)
            {
                Logger.WriteTraceInformation(result.Context, "Received delivery confirmation for: {0}", result);
                RaiseNotificationCompleted(result);
            }

            msg.Dispose();
        }

        /// <summary>
        /// Callback when an administration message is available.
        /// </summary>
        /// <param name="asyncResult">Context of async operation</param>
        private void OnAdministrationReceived(IAsyncResult asyncResult)
        {
            // Ensure that the service isn't shutdown until this operation completes
            if (!IncrementRefCount())
            {
                // Don't receive the message from the administration queue because the service is stopping
                return;
            }

            Message msg = null;

            try
            {
                // Extract the administration message
                msg = _administrationQueue.EndReceive(asyncResult);
                Debug.Assert(msg != null, "Received message should never be null");
            }
            catch (MessageQueueException ex)
            {
                // Ignore errors due to an empty queue as these are expected if this callback is invoked because of a timeout.
                if (ex.MessageQueueErrorCode != MessageQueueErrorCode.MessageNotFound && ex.MessageQueueErrorCode != MessageQueueErrorCode.IOTimeout)
                {
                    OperationalLogs.LogMessageQueueError(_administrationQueue.Path, ex);
                }
            }

            ProcessAdministrationMessage(msg);

            // Start listening for more administration messages
            _administrationQueue.BeginReceive(TimeSpan.FromMinutes(1), this, OnAdministrationReceived);

            DecrementRefCount();
        }

        /// <summary>
        /// Processes an administration message.
        /// </summary>
        /// <param name="msg">Administration message</param>
        private void ProcessAdministrationMessage(Message msg)
        {
            if (msg == null)
            {
                return;
            }

            Debug.Assert(msg.Acknowledgment != Acknowledgment.None && msg.Acknowledgment != Acknowledgment.ReachQueue && 
                msg.Acknowledgment != Acknowledgment.Receive, "Should only get negative acknowledgements");

            msg.Formatter = new BinaryMessageFormatter();

            NotificationItem notificationItem = msg.Body as NotificationItem;

            if (notificationItem != null)
            {
                Logger.WriteTraceError(notificationItem.Context, "Failed to deliver notification because of an MSMQ error. Error: {0}, Notification: {1}", msg.Acknowledgment, notificationItem);
                OperationalLogs.LogNotificationDeliveryError(_sendQueue.Path, msg.Acknowledgment.ToString());
                RaiseNotificationCompleted(new NotificationResult(notificationItem.Context, notificationItem.Id, NotificationResultStatus.Error, NotificationChannelType.None));
            }

            msg.Dispose();
        }

        /// <summary>
        /// Raises the NotificationReady event for the specified notification.
        /// </summary>
        /// <param name="notificationItem">Notification</param>
        private void RaiseNotificationReady(NotificationItem notificationItem)
        {
            Debug.Assert(notificationItem != null, "notificationItem != null");

            EventHandler<NotificationReadyEventArgs> notificationReady = NotificationReady;

            if (notificationReady != null)
            {
                notificationReady(this, new NotificationReadyEventArgs(notificationItem));
            }
        }

        /// <summary>
        /// Raises notification completed event for the specified notification.
        /// </summary>
        /// <param name="result">Notification result</param>
        private void RaiseNotificationCompleted(NotificationResult result)
        {
            Debug.Assert(result != null, "Result should always be specified");

            EventHandler<NotificationCompletedEventArgs> notificationCompleted = NotificationCompleted;

            if (notificationCompleted != null)
            {
                notificationCompleted(this, new NotificationCompletedEventArgs(result));
            }
        }

        /// <summary>
        /// Increment the count of current MSMQ-related operations.
        /// </summary>
        /// <remarks>
        /// The reference counting ensures that the service does not shutdown part way through an MSMQ-related operation.
        /// </remarks>
        /// <returns>
        /// true if the service is not stopping; false otherwise.
        /// </returns>
        private bool IncrementRefCount()
        {
            lock (_syncRoot)
            {
                if (_stopping)
                {
                    return false;
                }

                _refCount++;
            }

            return true;
        }

        /// <summary>
        /// Decrement the count of current MSMQ-related operations.
        /// </summary>
        /// <remarks>
        /// The reference counting ensures that the service does not shutdown part way through an MSMQ-related operation.
        /// </remarks>
        private void DecrementRefCount()
        {
            lock (_syncRoot)
            {
                Debug.Assert(_refCount > 0, "DecrementRefCount should not be called without having called IncrementRefCount first");

                _refCount--;

                if (_stopping)
                {
                    if (_refCount <= 0)
                    {
                        // The last outstanding MSMQ-related operation has completed, so unblock the shutdown process.
                        Debug.Assert(_stoppedEvent != null, "_stoppedEvent should not be null");
                        _stoppedEvent.Set();
                    }
                }
            }
        }
    }
}
